package cn.ztuo.bitrade.consumer;

import cn.ztuo.bitrade.constant.TransactionType;
import cn.ztuo.bitrade.dao.CoinDao;
import cn.ztuo.bitrade.dao.MemberTransactionDao;
import cn.ztuo.bitrade.dao.MemberWalletDao;
import cn.ztuo.bitrade.dao.WinningRecordDao;
import cn.ztuo.bitrade.entity.Coin;
import cn.ztuo.bitrade.entity.MemberTransaction;
import cn.ztuo.bitrade.entity.MemberWallet;
import cn.ztuo.bitrade.entity.WinningRecord;
import cn.ztuo.bitrade.util.BigDecimalUtils;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;
import org.springframework.transaction.annotation.Transactional;

import java.util.Date;
import java.util.List;

@Slf4j
@Component
public class AwardBonusConsumer {

    @Autowired
    private WinningRecordDao winningRecordDao;

    @Autowired
    private MemberWalletDao memberWalletDao;

    @Autowired
    private CoinDao coinDao;

    @Autowired
    private MemberTransactionDao memberTransactionDao;

    /**
     * 异步发放奖金
     * @param records
     */
    @KafkaListener(topics = "award-bonus", containerFactory = "kafkaListenerContainerFactory")
    @Transactional
    public void onAddCoinTradeByExchangeCoin(List<ConsumerRecord<String, String>> records) {
        for (int i = 0; i < records.size(); i++) {
            ConsumerRecord<String, String> record = records.get(i);
            log.info("topic={},accessKey={},value={}", record.topic(), record.key(), record.value());
            Long winningRecordId = Long.parseLong(record.key());
            WinningRecord winningRecord = winningRecordDao.findById(winningRecordId);
            // 查询中奖用户的USDT钱包
            Coin coin = coinDao.findByUnit("USDT");
            MemberWallet memberWallet = memberWalletDao.findByCoinAndMemberId(coin, winningRecord.getMember().getId());
            // 修改余额 = 余额 + 奖金
            memberWallet.setBalance(BigDecimalUtils.add(memberWallet.getBalance(), winningRecord.getAwardsSetting().getBonus()));
            memberWalletDao.save(memberWallet);
            // 中奖纪录状态修改为已发放
            winningRecord.setStatus(1);
            winningRecordDao.save(winningRecord);

            // 资产流水
            MemberTransaction memberTransaction = new MemberTransaction();
            memberTransaction.setMemberId(winningRecord.getMember().getId());
            memberTransaction.setSymbol("USDT");
            memberTransaction.setCreateTime(new Date());
            memberTransaction.setAmount(winningRecord.getAwardsSetting().getBonus());
            memberTransaction.setType(TransactionType.GIVE_OUT_BONUS);
            memberTransactionDao.save(memberTransaction);

            log.info("memberId为{},奖金已发放", winningRecord.getMember().getId());
        }
    }
}
